// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: jshin
#pragma once

#include <algorithm>
#include <map>
#include <memory>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include "coin/feed/fastfeed/feed.pb.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/executor/archive_executor.h"
#include "coin2/exchange/base/feed/system.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/feed_util/book_builder.h"
#include "coin2/exchange/base/feed_util/iv_greeks.h"
#include "coin2/exchange/base/feed_util/kline.h"
#include "coin2/exchange/base/feed_util/nav.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/di/constant.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/feed/fastfeed/fastfeed_reader.h"
#include "coin2/feed/universe/fastfeed_universe_reader.h"

namespace coin2::feed::fastfeed {
namespace impl {

using coin2::base::config::ValidateArchiveConfig;
using coin2::exchange::base::market::ExchangeType;
using coin2::exchange::base::market::MarketType;
using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::feed::ArchiveConfig;
using coin2::exchange::feed::FeedSystemConfig;
using ::fastfeed::proto::Feed;

// 1 FastFeedUniverseSubsystem = 1 mear
// so we can control root_dir, machine, etc per exchange.
class FastFeedUniverseSubsystem : public exchange::base::feed::IFeedSubsystem,
                                  public exchange::base::feed::FeedArchiveMixin {
 public:
  FastFeedUniverseSubsystem(
      const std::string& system_name,
      const MarketExchangeApiRecipe& mear,
      const coin2::exchange::di::ConstantInjection& ci,
      const FeedSystemConfig& config,
      exchange::base::executor::ArchiveSyncExecutor* ass,
      exchange::base::feed::IFeedSubscriber* callback)
      : IFeedSubsystem(system_name, mear.mea), FeedArchiveMixin(ass), mear_(mear), config_(config) {
    AddCallback(callback);
    CHECK_EQ_THROW(
        config.archive().archive_type(),
        coin2::exchange::feed::ArchiveConfig_ArchiveType_FASTFEED_UNIVERSE);
    ValidateArchiveConfig(config.archive());

    const auto& archive = config_.archive();
    auto interval = coin2::base::config::FromIntervalConfig(archive.interval());
    auto warmup = DurationFromString(archive.read_warmup_duration());

    std::set<std::string> universe_tags;
    std::string str_universe_tags;
    for (const auto& u : config_.universe_tags()) {
      universe_tags.insert(u);
      str_universe_tags += (u + ",");
    }

    LOG(INFO) << "[FastFeed Universe System]";
    LOG(INFO) << "              mea: " << mea_.String();
    LOG(INFO) << "    universe_tags: " << str_universe_tags;
    LOG(INFO) << "interval - warmup: " << DebugStringFromDuration(warmup);
    LOG(INFO) << "interval -  start: " << Iso8601FromTimestamp(interval.first);
    LOG(INFO) << "interval -    end: " << Iso8601FromTimestamp(interval.second);

    // FastFeedReader is capable of handling multiple exchanges,
    // but here we use it for one exchange
    start_time_ = interval.first;

    reader_ = std::make_unique<FastFeedUniverseReader>(
        mear_,
        config_.products(),
        archive.fastfeed_root_dir(),
        archive.machine(),
        universe_tags,
        warmup,
        interval.first,
        interval.second,
        archive.max_record_size());

    update_ptr_.reset(new FastFeedUpdate(system_name, mea_));
  }

  //     feed_type: Book,
  //     data_type: BookSnapshot,
  //     symbol: "ADA-ETH.Huobi.adaeth",

  void Prepare() override {}
  void Cleanup() override {}

  int64_t PeekTimestamp() override {
    if (!has_feed_) {
      while (reader_->Read(&feed_)) {
        feed_.set_timestamp(feed_.timestamp() + config_.archive().offset_ns());
        has_feed_ = true;
        break;
      }
      if (!has_feed_) {
        return 0;
      }
    }
    return feed_.timestamp();
  }

  void ConsumeRecord() override {
    using namespace ::fastfeed::proto;
    CHECK_THROW(has_feed_) << "Cannot consume a feed";
    has_feed_ = false;

    std::string norm_symbol;
    const std::string& fastfeed_symbol = feed_.symbol();
    int dots = std::count(fastfeed_symbol.begin(), fastfeed_symbol.end(), '.');

    // (BTC-USD or BTC-USD.xxx) 0~1 vs BTC-USD.20191206.Okex.BTC1206 2~3
    if (dots <= 1) {
      norm_symbol = fastfeed_symbol;
    } else {
      // coin1
      if (dots == 2) {
        norm_symbol = fastfeed_symbol.substr(0, fastfeed_symbol.find('.'));
      } else {
        if (dots == 3) {
          int dot2 = fastfeed_symbol.find('.', fastfeed_symbol.find('.') + 1);
          int dot3 = fastfeed_symbol.find('.', dot2 + 1);
          auto between = fastfeed_symbol.substr(dot2 + 1, dot3 - dot2 - 1);
          if (between == "P" || between == "C") {
            norm_symbol = fastfeed_symbol;
          } else {
            int dot = fastfeed_symbol.find('.', fastfeed_symbol.find('.') + 1);
            norm_symbol = fastfeed_symbol.substr(0, dot);
          }
        } else {
          int dot = fastfeed_symbol.find('.', fastfeed_symbol.find('.') + 1);
          norm_symbol = fastfeed_symbol.substr(0, dot);
        }
      }
    }

    FastFeedUpdate& update_ = *update_ptr_.get();
    update_.SetTimestamp(feed_.timestamp());
    // clear
    update_.SetKlines(nullptr);

    if (feed_.timestamp() >= start_time_) {
      if (products_.find(norm_symbol) == products_.end()) {
        products_[norm_symbol] = CreateProductFromNormString(mea_, norm_symbol, feed_.timestamp());
      }
      update_.SetProduct(*products_[norm_symbol]);
      update_.SetNativeSymbol(products_[norm_symbol]->feed_native_symbol());
    }

    const auto* pqmult = update_.GetPqMultipliers(norm_symbol);

    if (feed_.feed_type() == FEED_TYPE_METADATA) {
      if (feed_.data_type() == DATA_TYPE_BOOK_BUILDER_PARAMETER) {
        const auto& param = feed_.book_builder_param();
        update_.SetPriceQtyMultipliers(
            norm_symbol,
            pow(0.1, param.price_precision()),
            pow(0.1, param.qty_precision()));
      }
    } else if (feed_.feed_type() == FEED_TYPE_BOOK) {
      // not ready yet.
      if (!pqmult) {
        return;
      }
      BookBuilder* b = books_[norm_symbol].get();
      if (b == nullptr) {
        BookBuilder::BookOption book_type{BookBuilder::SNAPSHOT};
        if (config_.raw_parser().enable_diff_book()) {
          book_type = BookBuilder::DELTA;
        } else if (config_.raw_parser().publish_bbo_book_only()) {
          book_type = BookBuilder::LIGHT_BBO;
        }
        books_[norm_symbol].reset(b = new BookBuilder(book_type));
      }

      if (feed_.data_type() == DATA_TYPE_BOOK_SNAPSHOT) {
        const auto& snapshot = feed_.book_snapshot();
        auto tx = b->NewTransaction(feed_.origin_timestamp());
        tx.Clear();
        for (const auto& level : snapshot.ask()) {
          tx.UpdateAsk(
              level.price() * pqmult->price_multiplier,
              level.qty() * pqmult->qty_multiplier);
        }
        for (const auto& level : snapshot.bid()) {
          tx.UpdateBid(
              level.price() * pqmult->price_multiplier,
              level.qty() * pqmult->qty_multiplier);
        }
      } else if (feed_.data_type() == DATA_TYPE_BOOK_UPDATE) {
        const auto& diff = feed_.book_update();
        auto tx = b->NewTransaction(feed_.origin_timestamp());
        for (const auto& level : diff.ask()) {
          tx.UpdateAsk(
              level.price() * pqmult->price_multiplier,
              level.qty() * pqmult->qty_multiplier);
        }
        for (const auto& level : diff.bid()) {
          tx.UpdateBid(
              level.price() * pqmult->price_multiplier,
              level.qty() * pqmult->qty_multiplier);
        }
      }

      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      update_.SetFeedType(coin2::exchange::feed::BOOK_FEED);
      update_.SetBook(b);
      update_.MutableTs().set_exchange_publish_timestamp(b->Timestamp());
      onBookFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_TRADE) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& trade = feed_.trade();
      update_.SetFeedType(coin2::exchange::feed::TRADE_FEED);
      update_.SetTrade(&trade_);
      trade_ = {
          feed_.origin_timestamp(),
          static_cast<double>(trade.price() * pqmult->price_multiplier),
          static_cast<double>(trade.qty() * pqmult->qty_multiplier),
          ConvertTradeSide(trade.side()),
          feed_.feed_continue() == ::fastfeed::proto::FEED_CONTINUE_YES,
      };
      update_.MutableTs().set_exchange_publish_timestamp(trade_.timestamp);
      onTradeFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_LIQUIDATION_ORDER) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& liquidation = feed_.liquidation();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetLiquidationOrder(&liquidation_order_);
      coin::proto::OrderSide order_side = coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
      if (liquidation.direction() == ::fastfeed::proto::ORDER_DIRECTION_SELL) {
        order_side = coin::proto::OrderSide::SELL_ORDER;
      } else if (liquidation.direction() == ::fastfeed::proto::ORDER_DIRECTION_BUY) {
        order_side = coin::proto::OrderSide::BUY_ORDER;
      }
      liquidation_order_ = {
          feed_.origin_timestamp(),
          static_cast<double>(liquidation.price() * pqmult->price_multiplier),
          static_cast<double>(liquidation.qty() * pqmult->qty_multiplier),
          0,
          0,
          coin::proto::OrderType::UNKNOWN_ORDER_TYPE,
          order_side,
          coin::proto::OrderDuration::GTC_ORDER,
          coin::proto::OrderState::CREATED_ORDER,
          feed_.feed_continue() == ::fastfeed::proto::FEED_CONTINUE_YES,
          liquidation.order_id()
      };
      update_.MutableTs().set_exchange_publish_timestamp(liquidation_order_.timestamp);
      onLiquidationFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_OPEN_INTEREST) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& open_interest = feed_.open_interest();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetOpenInterest(&open_interest_);
      open_interest_ = {
          feed_.origin_timestamp(),
          static_cast<double>(open_interest.qty() * pqmult->qty_multiplier),
      };
      update_.MutableTs().set_exchange_publish_timestamp(open_interest_.timestamp);
      onOpenInterestFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_INDEX) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& index = feed_.index();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetIndex(&index_);
      index_ = {
          feed_.origin_timestamp(),
          static_cast<double>(index.price() * pqmult->price_multiplier),
      };
      update_.MutableTs().set_exchange_publish_timestamp(index_.timestamp);
      onIndexFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_MARK_PRICE) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& mark_price = feed_.mark_price();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetMarkPrice(&mark_price_);
      mark_price_ = {
          feed_.origin_timestamp(),
          static_cast<double>(mark_price.price() * pqmult->price_multiplier),
      };
      update_.MutableTs().set_exchange_publish_timestamp(mark_price_.timestamp);
      onMarkPriceFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_FUNDING_RATE) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& funding_rate = feed_.funding_rate();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetFundingRate(&funding_rate_);
      funding_rate_ = {
          feed_.origin_timestamp(),
          static_cast<double>(funding_rate.funding_rate() * kDefaultMultiplier),
          funding_rate.funding_time(),
          static_cast<double>(funding_rate.estimated_rate() * kDefaultMultiplier),
      };
      update_.MutableTs().set_exchange_publish_timestamp(funding_rate_.timestamp);
      onFundingRateFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_IV_GREEKS) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& iv_greeks = feed_.iv_greeks();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetIvGreeks(&iv_greeks_);
      iv_greeks_ = {
          feed_.origin_timestamp(),
          static_cast<double>(iv_greeks.ask_iv() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.bid_iv() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.mark_iv() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.delta() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.gamma() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.rho() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.theta() * kDefaultMultiplier),
          static_cast<double>(iv_greeks.vega() * kDefaultMultiplier),
      };
      update_.MutableTs().set_exchange_publish_timestamp(iv_greeks_.timestamp);
      onIvGreeksFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_NAV) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& nav = feed_.nav();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetNav(&nav_);

      std::vector<::UnderlyingAsset> basket;
      for (auto& ua : nav.basket()) {
        basket.emplace_back(
            std::string(ua.symbol_or_currency()),
            static_cast<double>(ua.amount() * kDefaultMultiplier));
      }
      nav_ = {
          feed_.origin_timestamp(),
          static_cast<double>(nav.actual_leverage() * kDefaultMultiplier),
          static_cast<double>(nav.nav() * pqmult->price_multiplier),
          static_cast<double>(nav.outstanding() * kDefaultMultiplier),
          basket,
      };
      update_.MutableTs().set_exchange_publish_timestamp(nav_.timestamp);
      onNavFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_KLINE) {
      if (!pqmult) {
        return;
      }
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      Klines* klines = nullptr;
      if (feed_.kline_snapshot().kline_size() > 0) {
        for (const auto& kline : feed_.kline_snapshot().kline()) {
          if (klines == nullptr) {
            klines = &(klines_[std::make_pair(norm_symbol, kline.interval())]);
            update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
            update_.SetKlines(klines);
            update_.MutableTs().set_exchange_publish_timestamp(feed_.origin_timestamp());
            klines->SnapshotClear();
          }

          // fastfeed only save freezed kline
          auto* pKline = klines->AddKline();
          pKline->timestamp = feed_.origin_timestamp();
          pKline->interval = kline.interval();
          pKline->open_timestamp = kline.open_timestamp();
          pKline->close_timestamp = kline.close_timestamp(), pKline->open = kline.open();
          pKline->close = kline.close();
          pKline->high = kline.high();
          pKline->low = kline.low();
          pKline->volume = kline.volume();
        }
      } else if (feed_.kline_update().kline_size() > 0) {
        for (const auto& kline : feed_.kline_update().kline()) {
          if (klines == nullptr) {
            klines = &(klines_[std::make_pair(norm_symbol, kline.interval())]);
            update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
            update_.SetKlines(klines);
            update_.MutableTs().set_exchange_publish_timestamp(feed_.origin_timestamp());
            klines->UpdateClear();
          }

          // fastfeed only save freezed kline
          auto* pKline = klines->AddKline();
          pKline->timestamp = feed_.origin_timestamp();
          pKline->interval = kline.interval();
          pKline->open_timestamp = kline.open_timestamp();
          pKline->close_timestamp = kline.close_timestamp(), pKline->open = kline.open();
          pKline->close = kline.close();
          pKline->high = kline.high();
          pKline->low = kline.low();
          pKline->volume = kline.volume();
        }
      }

      onKlineFeed(update_);
    }
  }

  std::vector<IProduct*> GetProducts() const override {
    std::vector<IProduct*> out;
    for (const auto& p : reader_->products()) {
      out.push_back(p);
    }
    return out;
  }

 private:
  const MarketExchangeApiRecipe mear_;
  const FeedSystemConfig& config_;

  // states
  int64_t start_time_;
  std::unique_ptr<FastFeedUniverseReader> reader_;
  bool has_feed_ = false;
  std::map<std::string, std::unique_ptr<BookBuilder>> books_;
  Trade trade_{};
  Feed feed_;
  std::unordered_map<int, std::string> index_to_norm_symbol_;
  LiquidationOrder liquidation_order_{};
  OpenInterest open_interest_{};
  Index index_{};
  MarkPrice mark_price_{};
  FundingRate funding_rate_{};
  IvGreeks iv_greeks_{};
  Nav nav_{};
  // <symbol, interval> => klines
  std::map<std::pair<std::string, ::fastfeed::proto::KlineInterval>, Klines> klines_;
  std::unique_ptr<FastFeedUpdate> update_ptr_;

  // cache
  // norm symbol => product
  std::map<std::string, std::unique_ptr<IProduct>> products_;
};
}  // namespace impl
using impl::FastFeedUniverseSubsystem;
}  // namespace coin2::feed::fastfeed